package com.ykht.proj.activity.mq.consumer;
import com.ykht.proj.activity.handle.OrderHandler;
import com.ykht.proj.activity.mq.MqConfig;
import com.ykht.proj.activity.service.CnActivityOrderService;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

@Data
@Slf4j
@Component
public class ScheduledMsgOrderConsumer {
	@Resource
	private CnActivityOrderService cnActivityOrderService;

	@Resource
	private OrderHandler orderHandler;

	@Resource
	private MqConfig mqConfig;

	@PostConstruct
	public void order(){
		DefaultMQPushConsumer consumer = null;
		try {
			// 实例化消费者
			consumer = new DefaultMQPushConsumer(mqConfig.getConsumerGroup());
			// 设置NameServer的地址
			consumer.setNamesrvAddr(mqConfig.getNamasrv());
			// 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
			consumer.subscribe(mqConfig.getTopic(), "*");
			//设置集群消费
    	    consumer.setMessageModel(MessageModel.CLUSTERING);
			// 注册回调实现类来处理从broker拉取回来的消息
			consumer.registerMessageListener((MessageListenerConcurrently)(msgList, context)->{
				try {
					msgList.stream().forEach(msg->{
						log.info("消费消息:{}",msg.toString());
						String id = new String(msg.getBody());
						orderHandler.expireOrder(Long.parseLong(id));
					});
					return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
				} catch (Exception e) {
					e.printStackTrace();
					return ConsumeConcurrentlyStatus.RECONSUME_LATER;
				}
			});
			// 启动消费者实例
			consumer.start();
			log.info("启动消费者：{}",consumer.toString());
		} catch (MQClientException e) {
			e.printStackTrace();
			consumer.shutdown();
		}
	}

}